﻿// =====================================================
// Copyright (C) 2016-2018 Fanjia
// All rights reserved
// 机器名称: PC-LIWEI
// 创建时间: 2018/10/22 20:10:15
// 创 建 人: 李伟
// ======================================================
using System;
using System.Collections.Generic;
using System.Configuration;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Nest;
using WindowsFormsApplication1.Common;

namespace JobBatDeal.Common
{
    /// <summary>
    /// ElasticSearchHelper
    /// </summary>
    public class ElasticSearchHelper
    {
        private static ElasticClient _client;

        public static string IndexName;

        public static ElasticClient GetClient()
        {
            if (_client == null)
            {
                IndexName = ConfigurationManager.AppSettings["EsDefaultIndex"].ToLower();

                var esUrl = ConfigurationManager.AppSettings["EsUrl"];
                var node = new Uri(esUrl);
                var settings = new ConnectionSettings(node).DefaultIndex(IndexName);

                var esUser = ConfigurationManager.AppSettings["EsUser"];
                if (!string.IsNullOrEmpty(esUser))
                {
                    var esPwd = ConfigurationManager.AppSettings["EsPwd"];
                    settings.BasicAuthentication(esUser, esPwd);
                }                

                _client = new ElasticClient(settings);
            }

            return _client;
        }

        public static ElasticClient GetClient(string indexName)
        {
            if (_client == null)
            {
                var esUrl = ConfigurationManager.AppSettings["EsUrl"];
                var node = new Uri(esUrl);
                var settings = new ConnectionSettings(node).DefaultIndex(indexName);

                var esUser = ConfigurationManager.AppSettings["EsUser"];
                if (!string.IsNullOrEmpty(esUser))
                {
                    var esPwd = ConfigurationManager.AppSettings["EsPwd"];
                    settings.BasicAuthentication(esUser, esPwd);
                }

                _client = new ElasticClient(settings);
            }

            return _client;
        }

        /// <summary>
        /// index 是否存在
        /// </summary>
        /// <param name="index"></param>
        /// <returns></returns>
        public static bool IndexExists(string index)
        {
            var res = GetClient().IndexExists(index);
            return res.Exists;
        }


        /// <summary>
        /// 创建index
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="index"></param>
        /// <returns></returns>
        public static bool InitIndex<T>(string index) where T : class
        {
            if (!IndexExists(index))
            {
                GetClient().InitializeIndexMap<T>(index);
            }
            return true;
        }

        /// <summary>
        /// 创建index 并新增数据
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="entity"></param>
        /// <param name="index"></param>
        /// <returns></returns>
        public static bool Index<T>(T entity, string index) where T : class
        {
            if (!IndexExists(index))
            {
                GetClient().InitializeIndexMap<T>(index);
            }
            var response = GetClient().Index(entity,
                s => s.Index(index));

            if (!response.IsValid)
                throw new Exception("新增数据失败:" + response.OriginalException.Message);

            return true;
        }

        /// <summary>
        /// 批量新增
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="entity"></param>
        /// <param name="index"></param>
        public static bool BulkIndex<T>(List<T> entity, string index) where T : class
        {
            if (!IndexExists(index))
            {
                GetClient().InitializeIndexMap<T>(index);
            }

            //var bulkRequest = new BulkRequest(index)
            //{
            //    Operations = new List<IBulkOperation>()
            //};
            //var idxops = entity.Select(o => new BulkIndexOperation<T>(o)).Cast<IBulkOperation>().ToList();
            //bulkRequest.Operations = idxops;
            //var response = GetClient().Bulk(bulkRequest);

            const int size = 1000;
            var tokenSource = new CancellationTokenSource();

            var observableBulk = GetClient().BulkAll(entity, f => f
                    .MaxDegreeOfParallelism(8)
                    .BackOffTime(TimeSpan.FromSeconds(10))
                    .BackOffRetries(2)
                    .Size(size)
                    .RefreshOnCompleted()
                    .Index(index)
                    .BufferToBulk((r, buffer) => r.IndexMany(buffer))
                , tokenSource.Token);

            var countdownEvent = new CountdownEvent(1);

            Exception exception = null;

            var bulkAllObserver = new BulkAllObserver(
                onNext: response =>
                {
                    //LogHelper.Logger.Info($"Indexed {response.Page * size} with {response.Retries} retries");
                },
                onError: ex =>
                {
                    //LogHelper.Logger.Info("BulkAll Error", ex);
                    exception = ex;
                    countdownEvent.Signal();
                },
                onCompleted: () =>
                {
                    //LogHelper.Logger.Info("BulkAll Finished");
                    countdownEvent.Signal();
                });

            observableBulk.Subscribe(bulkAllObserver);

            countdownEvent.Wait(tokenSource.Token);

            if (exception != null)
            {
                throw new Exception("新增数据失败:" + exception.Message, exception);
            }
            else
            {
                return true;
            }

            //if (!response.IsValid)
            //    throw new Exception("新增数据失败:" + response.OriginalException.Message);

            //return true;
        }

        /// <summary>
        /// 删除index
        /// </summary>
        /// <param name="index"></param>
        public static bool RemoveIndex(string index)
        {
            if (!IndexExists(index)) return false;
            var response = GetClient().DeleteIndex(index);

            if (!response.IsValid)
                throw new Exception("删除index失败:" + response.OriginalException.Message);

            return true;
        }

        /// <summary>
        /// Alias 别名操作
        /// </summary>
        /// <param name="aliasRequest"></param>
        public static IBulkAliasResponse Alias(IBulkAliasRequest aliasRequest)
        {
            var response = GetClient().Alias(aliasRequest);

            if (!response.IsValid)
                throw new Exception("操作Alias失败:" + response.OriginalException.Message);
            return response;
        }

        /// <summary>
        /// 全文检索
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="searchRequest"></param>
        /// <returns></returns>
        public static ISearchResponse<T> SearchPage<T>
            (SearchRequest<T> searchRequest) where T : class
        {
            var response = GetClient().Search<T>(searchRequest);
            if (!response.IsValid)
                throw new Exception("查询失败:" +
                                                 response.OriginalException.Message);
            return response;
        }

        /// <summary>
        /// 查询指定数据
        /// </summary>
        /// <param name="request"></param>
        public static T Get<T>(GetRequest request) where T : class
        {
            var response = GetClient().Get<T>(request);
            if (!response.IsValid)
                throw new Exception("查询失败:" +
                                                 response.OriginalException.Message);
            return response.Source;
        }

        /// <summary>
        /// 删除指定数据
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="deleteRequest"></param>
        /// <returns></returns>
        public static IDeleteByQueryResponse DeleteByQuery<T>(DeleteByQueryRequest<T> deleteRequest) where T : class
        {
            var response = GetClient().DeleteByQuery(deleteRequest);
            if (!response.IsValid)
                throw new Exception("删除失败:" +
                                                 response.OriginalException.Message);
            return response;
        }

        /// <summary>
        /// 删除指定数据 指定索引库
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="deleteRequest"></param>
        /// <returns></returns>
        public static IDeleteByQueryResponse DeleteByQuery<T>(DeleteByQueryRequest<T> deleteRequest, string indexName) where T : class
        {
            var response = GetClient(indexName).DeleteByQuery(deleteRequest);
            if (!response.IsValid)
                throw new Exception("删除失败:" +
                                                 response.OriginalException.Message);
            return response;
        }

        /// <summary>
        /// 删除指定数据
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="selector"></param>
        /// <returns></returns>
        public static long DeleteByQuery<T>(Func<DeleteByQueryDescriptor<T>, IDeleteByQueryRequest> selector) where T : class
        {
            var deleteResponse = GetClient().DeleteByQuery(selector);

            return deleteResponse.Deleted;
        }

        /// <summary>
        /// 删除指定数据
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="deletePath"></param>
        /// <returns></returns>
        public static bool DeleteByDocumentPath<T>(DocumentPath<T> deletePath) where T : class
        {
            var response = GetClient().Delete(deletePath);
            if (!response.IsValid)
                throw new Exception("删除失败:" +
                                    response.OriginalException.Message);
            return true;
        }

        /// <summary>
        /// 删除指定数据
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="deletePath"></param>
        /// <returns></returns>
        public static bool DeleteByDocumentPathN<T>(DocumentPath<T> deletePath) where T : class
        {
            var response = GetClient().Delete(deletePath);
            return response.IsValid;
        }


        /// <summary>
        /// 更新操作
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="request"></param>
        /// <returns></returns>
        public static IUpdateResponse<T> Update<T>(IUpdateRequest<T, object> request) where T : class
        {
            var response = GetClient().Update(request);
            if (!response.IsValid)
                throw new Exception("更新失败:" +
                                    response.OriginalException.Message);
            return response;
        }

        /// <summary>
        /// 更新
        /// </summary>
        /// <typeparam name="T1"></typeparam>
        /// <typeparam name="T2"></typeparam>
        /// <param name="request"></param>
        /// <returns></returns>
        public static IUpdateResponse<T1> Update<T1, T2>(IUpdateRequest<T1, T2> request) where T1 : class where T2 : class
        {
            var response = GetClient().Update(request);
            if (!response.IsValid)
                throw new Exception("更新失败:" + response.OriginalException.Message);
            return response;
        }

        public static bool BulkAll<T>(IEnumerable<T> list) where T : class
        {
            const int size = 1000;
            var tokenSource = new CancellationTokenSource();

            var observableBulk = GetClient().BulkAll(list, f => f
                    .MaxDegreeOfParallelism(8)
                    .BackOffTime(TimeSpan.FromSeconds(10))
                    .BackOffRetries(2)
                    .Size(size)
                    .RefreshOnCompleted()
                    .Index(IndexName)
                    .BufferToBulk((r, buffer) => r.IndexMany(buffer))
                , tokenSource.Token);

            var countdownEvent = new CountdownEvent(1);

            Exception exception = null;

            var bulkAllObserver = new BulkAllObserver(
                onNext: response =>
                {
                    LogHelper.logInfor.Info($"Indexed {response.Page * size} with {response.Retries} retries");
                },
                onError: ex =>
                {
                    LogHelper.logError.Info("BulkAll Error", ex);
                    exception = ex;
                    countdownEvent.Signal();
                },
                onCompleted: () =>
                {
                    LogHelper.logInfor.Info("BulkAll Finished");
                    countdownEvent.Signal();
                });

            observableBulk.Subscribe(bulkAllObserver);

            countdownEvent.Wait(tokenSource.Token);

            if (exception == null)
            {
                return true;
            }

            return false;
        }
    }
}
